/*
 * JobService.java
 *
 * Copyright 2012 the original author or authors(ninglong).
 *
 * Licensed under the org.geek.job.service, Version 1.0 (the "License");
 *
 */
package org.geek.app.job.service.impl;

import java.io.IOException;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.FutureTask;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import javax.annotation.Resource;

import net.sf.ehcache.util.NamedThreadFactory;

import org.geek.app.job.service.IJobService;
import org.geek.core.mybatis.domain.Job;
import org.geek.core.mybatis.mapper.JobMapper;
import org.springframework.stereotype.Service;

/**
 * <p></p>
 * @author ninglong bhd2973@qq.com
 * @version V1.0
 */
@Service
public class JobService implements IJobService{
	
	private static final String THREAD_NAME_JOB = "JOB_PREFIX";
	
	private ConcurrentHashMap<String, FutureTask<Boolean>> tasks = new ConcurrentHashMap<String, FutureTask<Boolean>>();
	private ConcurrentHashMap<FutureTask<Boolean>, Job> jobs = new ConcurrentHashMap<FutureTask<Boolean>, Job>();
	
	private ScheduledThreadPoolExecutor exec = null;
	
	private int poolSize = Runtime.getRuntime().availableProcessors()*2+1;
	
	private int retryCount = 3;
	
	private int retryTimePeriod = 180;
	
	private boolean start = false;
	
	@Resource
	private JobMapper jobMapper;
	
	@Override
	public void start() {
		if(start)return;
		initPool();
		start = true;
	}
	private void initPool(){
		exec = new ScheduledThreadPoolExecutor(poolSize,new NamedThreadFactory(THREAD_NAME_JOB)){
			@SuppressWarnings("unchecked")
			protected void afterExecute(Runnable r,Throwable t){
				try{
					FutureTask<Boolean> task = (FutureTask<Boolean>)r;
					try{
						task.get();
					}catch(Throwable e){
						t = e;
					}
					if(t != null){
						t = getRootCause(t);
					}
					Job job = jobs.remove(task);
					if(job != null){
						String key = job.getKey();
						tasks.remove(key);
						if(t !=null || job.getErrorDesc() !=null){
							if(job.isErrorJob()){
								//添加错误的job信息
								job.setErrorDesc("error");
								jobMapper.updateJob(job);
								//删除已执行的job
								jobMapper.delJob(job.getJobId());
								return;
							}
							if(t instanceof IOException && job.getRetryCount() <retryCount){
								System.out.println("重试");
								job.setJobExceuteTime(System.currentTimeMillis()+retryTimePeriod*1000);
								job.setRetryCount(job.getRetryCount()+1);
								job.setJobRun(false);
								startJob(job);
							}else{
								String errorDesc = null;
								if(t != null){
									errorDesc = t.getMessage();
								}else{
									errorDesc = job.getErrorDesc();
								}
								if(job.getJobType() == Job.JOB_DB){
									System.out.println("不再执行，下次启再执行");
								}
								if(errorDesc !=null){
									if(errorDesc.length() > 100){
										errorDesc = errorDesc.substring(0,100);
									}else{
										errorDesc = "error";
									}
									job.setErrorDesc(errorDesc);
									jobMapper.updateJob(job);
								}else{
									System.out.println("不再执行，不再执行");
								}
							}
						}else{
							if(job.getCronExpression() != null){
								job.setRetryCount(0);
								job.setJobRun(false);
								startJob(job);
							}else{
								if(job.getJobType() == Job.JOB_DB && job.getJobId()>0){
									jobMapper.delJob(job.getJobId());
								}
							}
						}
					}
				}catch(Exception e){
					e.printStackTrace();
				}
			}
		};
		exec.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardPolicy());
	}
	@Override
	public void loadSerialJob() {
		long now = System.currentTimeMillis();
		int timeoutJobNum = 0;
		List<Job> list = jobMapper.getJobList();
		for(Job job :list){
			try {
				if(job.getErrorDesc() !=null){
					job.setErrorJob(true);
				}
				if(job.getCronExpression() ==null){
					if(job.getJobExceuteTime() - now <=0){
						timeoutJobNum++;
					}
				}
				this.startJob(job);
			} catch (Exception e) {
				int jobId = job.getJobId();
				if(job.isErrorJob()){
					job.setErrorDesc("error");
					jobMapper.updateJob(job);
					jobMapper.delJob(jobId);
				}else{
					String errorDesc = e.getMessage();
					if(errorDesc !=null){
						errorDesc = errorDesc.substring(0,100);
					}else{
						errorDesc = "error";
					}
					job.setErrorDesc(errorDesc);
					jobMapper.updateJob(job);
				}
			}
		}
		System.out.println("超时job"+timeoutJobNum);
	}

	public void destory(){
		exec.shutdownNow();
		start = false;
	}
	@SuppressWarnings("unchecked")
	@Override
	public boolean startJob(Job job) {
		if(job ==null)return false;
		if(!start)this.start();
		String key = job.getKey();
		if(tasks.containsKey(key)){
			delJob(job.getJobGroupName(), job.getJobIdInGroup());
		}
		if(job.getJobType() == Job.JOB_DB && job.getJobId()>0){
			jobMapper.insertJob(job);
		}
		long delay = 0;
		if(job.getCronExpression() !=null & job.getRetryCount() == 0){
			delay = job.getNextTime();
			if(delay == -1){
				if(job.getJobType() == Job.JOB_DB && job.getJobId()>0){
					jobMapper.delJob(job.getJobId());
					return false;
				}
			}
		}else{
			delay = job.getJobExceuteTime() - System.currentTimeMillis();
		}
		FutureTask<Boolean> futureTask = (FutureTask<Boolean>)exec.schedule(job, delay, TimeUnit.MILLISECONDS);
		tasks.put(key, futureTask);
		jobs.put(futureTask, job);
		return true;
	}

	@Override
	public boolean delJob(String groupName, String groupId) {
		String key = new StringBuilder(groupName).append("_").append(groupId).toString();
		FutureTask<Boolean> futureTask = tasks.remove(key);
		if(futureTask !=null){
			exec.remove(futureTask);
			Job job = jobs.remove(futureTask);
			if(job!=null){
				if(!job.isJobRun()){
					job.setJobRun(true);
					job.setRepeatRunRule(null);
				}
				if(job.getJobType() == Job.JOB_DB && job.getJobId()>0){
					jobMapper.delJob(job.getJobId());
				}
			}
			return true;
		}
		return false;
	}

	@Override
	public boolean hasJob(String jobKey) {
		return tasks.containsKey(jobKey);
	}
	
	private Throwable getRootCause(Throwable t){
		while(t.getCause() != null){
			t = t.getCause();
		}
		return t;
	}

	public void setPoolSize(int poolSize) {
		this.poolSize = poolSize;
	}

	public void setRetryCount(int retryCount) {
		this.retryCount = retryCount;
	}

	public void setRetryTimePeriod(int retryTimePeriod) {
		this.retryTimePeriod = retryTimePeriod;
	}

}
